/**
 * @file simple_pair.c
 * @brief nanomsg PAIR 模式演示
 * 
 * 这个示例展示了如何使用 nanomsg 的 PAIR 模式进行双向通信。
 * PAIR 模式提供了一对一的双向通信通道，适用于线程间或进程间通信。
 */

#include "common.h"

// 全局变量定义
volatile int g_running = 1;
stats_t g_stats = {0};

/**
 * @brief 节点1的实现 - 发送方
 */
int run_node1(const char* address) {
    int sock;
    char buffer[BUFFER_SIZE];
    message_t* msg;
    int serialized_len;
    int counter = 0;
    
    log_message(LOG_INFO, "Starting Node1 (Sender) with address: %s", address);
    
    // 创建 PAIR 套接字
    sock = nn_socket(AF_SP, NN_PAIR);
    CHECK_SOCKET(sock);
    
    // 绑定地址
    int bind_result = nn_bind(sock, address);
    CHECK_BIND(bind_result, address);
    
    log_message(LOG_INFO, "Node1 bound to %s, waiting for peer...", address);
    
    // 等待连接建立
    sleep(1);
    
    while (should_continue()) {
        // 创建消息
        snprintf(buffer, sizeof(buffer), "Hello from Node1, message #%d", ++counter);
        msg = create_message(MSG_TEXT, buffer, strlen(buffer));
        
        // 序列化消息
        char send_buffer[MAX_MESSAGE_SIZE];
        serialized_len = serialize_message(msg, send_buffer, sizeof(send_buffer));
        
        if (serialized_len > 0) {
            // 发送消息
            int bytes_sent = safe_send(sock, send_buffer, serialized_len, 0);
            if (bytes_sent > 0) {
                update_send_stats(bytes_sent);
                log_message(LOG_INFO, "Sent: %s", buffer);
            } else {
                log_message(LOG_ERROR, "Failed to send message: %s", nn_strerror(nn_errno()));
                update_error_stats();
            }
        }
        
        destroy_message(msg);
        
        // 尝试接收回复
        int bytes_received = safe_recv(sock, buffer, sizeof(buffer), NN_DONTWAIT);
        if (bytes_received > 0) {
            message_t* recv_msg = deserialize_message(buffer, bytes_received);
            if (recv_msg) {
                update_recv_stats(bytes_received);
                log_message(LOG_INFO, "Received: %s", recv_msg->data);
                destroy_message(recv_msg);
            }
        } else if (nn_errno() != EAGAIN) {
            log_message(LOG_WARN, "Receive error: %s", nn_strerror(nn_errno()));
        }
        
        sleep(2);  // 2秒间隔
    }
    
    nn_close(sock);
    return 0;
}

/**
 * @brief 节点2的实现 - 接收方
 */
int run_node2(const char* address) {
    int sock;
    char buffer[BUFFER_SIZE];
    message_t* msg;
    int serialized_len;
    int counter = 0;
    
    log_message(LOG_INFO, "Starting Node2 (Receiver) with address: %s", address);
    
    // 创建 PAIR 套接字
    sock = nn_socket(AF_SP, NN_PAIR);
    CHECK_SOCKET(sock);
    
    // 连接到对端
    int connect_result = nn_connect(sock, address);
    CHECK_CONNECT(connect_result, address);
    
    log_message(LOG_INFO, "Node2 connected to %s", address);
    
    while (should_continue()) {
        // 接收消息
        int bytes_received = safe_recv(sock, buffer, sizeof(buffer), 0);
        if (bytes_received > 0) {
            message_t* recv_msg = deserialize_message(buffer, bytes_received);
            if (recv_msg) {
                update_recv_stats(bytes_received);
                log_message(LOG_INFO, "Received: %s", recv_msg->data);
                
                // 发送回复
                snprintf(buffer, sizeof(buffer), "ACK from Node2 for message #%d", ++counter);
                msg = create_message(MSG_TEXT, buffer, strlen(buffer));
                
                char send_buffer[MAX_MESSAGE_SIZE];
                serialized_len = serialize_message(msg, send_buffer, sizeof(send_buffer));
                
                if (serialized_len > 0) {
                    int bytes_sent = safe_send(sock, send_buffer, serialized_len, 0);
                    if (bytes_sent > 0) {
                        update_send_stats(bytes_sent);
                        log_message(LOG_INFO, "Sent ACK: %s", buffer);
                    } else {
                        log_message(LOG_ERROR, "Failed to send ACK: %s", nn_strerror(nn_errno()));
                        update_error_stats();
                    }
                }
                
                destroy_message(msg);
                destroy_message(recv_msg);
            }
        } else if (bytes_received < 0) {
            if (nn_errno() == ETERM || nn_errno() == EINTR) {
                break;  // 正常退出
            }
            log_message(LOG_ERROR, "Receive error: %s", nn_strerror(nn_errno()));
            update_error_stats();
        }
    }
    
    nn_close(sock);
    return 0;
}

/**
 * @brief 打印使用说明
 */
void print_pair_usage(const char* program_name) {
    printf("Usage: %s <node>\n", program_name);
    printf("  node: 'node1' or 'node2'\n");
    printf("\n");
    printf("Example:\n");
    printf("  Terminal 1: %s node1\n", program_name);
    printf("  Terminal 2: %s node2\n", program_name);
    printf("\n");
    printf("Options:\n");
    printf("  -h, --help     Show this help message\n");
    printf("  -v, --version  Show version information\n");
    printf("  -a, --address  Specify custom address (default: %s)\n", DEFAULT_TCP_ADDR);
    printf("\n");
    printf("Supported address formats:\n");
    printf("  tcp://ip:port     - TCP connection\n");
    printf("  ipc://path        - IPC connection\n");
    printf("  inproc://name     - In-process connection\n");
}

/**
 * @brief 主函数
 */
int main(int argc, char* argv[]) {
    const char* address = DEFAULT_TCP_ADDR;
    
    // 检查参数
    if (argc < 2) {
        print_pair_usage(argv[0]);
        return 1;
    }
    
    // 解析命令行参数
    for (int i = 1; i < argc; i++) {
        if (strcmp(argv[i], "-h") == 0 || strcmp(argv[i], "--help") == 0) {
            print_pair_usage(argv[0]);
            return 0;
        } else if (strcmp(argv[i], "-v") == 0 || strcmp(argv[i], "--version") == 0) {
            print_nanomsg_info();
            return 0;
        } else if (strcmp(argv[i], "-a") == 0 || strcmp(argv[i], "--address") == 0) {
            if (i + 1 < argc) {
                address = argv[++i];
            } else {
                log_message(LOG_ERROR, "Option %s requires an argument", argv[i]);
                return 1;
            }
        }
    }
    
    // 初始化
    setup_signal_handlers();
    init_stats();
    check_nanomsg_version();
    
    log_message(LOG_INFO, "=== Nanomsg PAIR Mode Demo ===");
    log_message(LOG_INFO, "Address type: %s", parse_address_type(address));
    log_message(LOG_INFO, "Press Ctrl+C to stop");
    
    int result = 0;
    
    // 根据节点类型运行
    if (strcmp(argv[argc-1], "node1") == 0) {
        result = run_node1(address);
    } else if (strcmp(argv[argc-1], "node2") == 0) {
        result = run_node2(address);
    } else {
        log_message(LOG_ERROR, "Invalid node type: %s", argv[argc-1]);
        print_pair_usage(argv[0]);
        return 1;
    }
    
    // 打印统计信息
    log_message(LOG_INFO, "Shutting down...");
    print_stats();
    
    return result;
}

// 实现公共函数（简化版本，完整版本应该在单独的 common.c 文件中）

void init_stats(void) {
    memset(&g_stats, 0, sizeof(g_stats));
    g_stats.start_time = time(NULL);
}

void print_stats(void) {
    time_t duration = time(NULL) - g_stats.start_time;
    printf("\n" COLOR_CYAN "=== Statistics ===" COLOR_RESET "\n");
    printf("Runtime: %ld seconds\n", duration);
    printf("Messages sent: %lu\n", g_stats.messages_sent);
    printf("Messages received: %lu\n", g_stats.messages_received);
    printf("Bytes sent: %lu\n", g_stats.bytes_sent);
    printf("Bytes received: %lu\n", g_stats.bytes_received);
    printf("Errors: %u\n", g_stats.errors);
    if (duration > 0) {
        printf("Send rate: %.2f msg/s\n", (double)g_stats.messages_sent / duration);
        printf("Receive rate: %.2f msg/s\n", (double)g_stats.messages_received / duration);
    }
}

void log_message(log_level_t level, const char* format, ...) {
    const char* level_colors[] = {
        COLOR_BLUE,    // DEBUG
        COLOR_GREEN,   // INFO
        COLOR_YELLOW,  // WARN
        COLOR_RED      // ERROR
    };
    
    const char* level_names[] = {"DEBUG", "INFO", "WARN", "ERROR"};
    
    char timestamp[32];
    get_timestamp_string(timestamp, sizeof(timestamp));
    
    va_list args;
    va_start(args, format);
    
    printf("%s[%s] %s%s" COLOR_RESET " ", 
           level_colors[level], timestamp, level_names[level], COLOR_RESET);
    vprintf(format, args);
    printf("\n");
    
    va_end(args);
    fflush(stdout);
}

void handle_error(const char* message) {
    log_message(LOG_ERROR, "%s: %s", message, nn_strerror(nn_errno()));
    update_error_stats();
}

void signal_handler(int sig) {
    if (sig == SIGINT || sig == SIGTERM) {
        log_message(LOG_INFO, "Received signal %d, shutting down gracefully...", sig);
        g_running = 0;
    }
}

void setup_signal_handlers(void) {
    signal(SIGINT, signal_handler);
    signal(SIGTERM, signal_handler);
}

message_t* create_message(message_type_t type, const char* data, uint32_t length) {
    static uint32_t sequence = 0;
    
    message_t* msg = malloc(sizeof(message_t));
    if (!msg) return NULL;
    
    msg->type = type;
    msg->sequence = ++sequence;
    msg->timestamp = get_timestamp();
    msg->data_length = length;
    
    if (length > 0 && data) {
        memcpy(msg->data, data, length);
    }
    
    return msg;
}

void destroy_message(message_t* msg) {
    if (msg) {
        free(msg);
    }
}

int serialize_message(const message_t* msg, char* buffer, size_t buffer_size) {
    if (!msg || !buffer || buffer_size < sizeof(message_t)) {
        return -1;
    }
    
    size_t total_size = sizeof(message_type_t) + sizeof(uint32_t) * 3 + msg->data_length;
    if (total_size > buffer_size) {
        return -1;
    }
    
    char* ptr = buffer;
    memcpy(ptr, &msg->type, sizeof(msg->type)); ptr += sizeof(msg->type);
    memcpy(ptr, &msg->sequence, sizeof(msg->sequence)); ptr += sizeof(msg->sequence);
    memcpy(ptr, &msg->timestamp, sizeof(msg->timestamp)); ptr += sizeof(msg->timestamp);
    memcpy(ptr, &msg->data_length, sizeof(msg->data_length)); ptr += sizeof(msg->data_length);
    memcpy(ptr, msg->data, msg->data_length);
    
    return total_size;
}

message_t* deserialize_message(const char* buffer, size_t length) {
    if (!buffer || length < sizeof(message_type_t) + sizeof(uint32_t) * 3) {
        return NULL;
    }
    
    message_t* msg = malloc(sizeof(message_t));
    if (!msg) return NULL;
    
    const char* ptr = buffer;
    memcpy(&msg->type, ptr, sizeof(msg->type)); ptr += sizeof(msg->type);
    memcpy(&msg->sequence, ptr, sizeof(msg->sequence)); ptr += sizeof(msg->sequence);
    memcpy(&msg->timestamp, ptr, sizeof(msg->timestamp)); ptr += sizeof(msg->timestamp);
    memcpy(&msg->data_length, ptr, sizeof(msg->data_length)); ptr += sizeof(msg->data_length);
    
    if (msg->data_length > MAX_MESSAGE_SIZE || 
        length < sizeof(message_type_t) + sizeof(uint32_t) * 3 + msg->data_length) {
        free(msg);
        return NULL;
    }
    
    memcpy(msg->data, ptr, msg->data_length);
    
    return msg;
}

uint32_t get_timestamp(void) {
    return (uint32_t)time(NULL);
}

const char* get_timestamp_string(char* buffer, size_t size) {
    time_t now = time(NULL);
    struct tm* tm_info = localtime(&now);
    strftime(buffer, size, "%H:%M:%S", tm_info);
    return buffer;
}

const char* parse_address_type(const char* addr) {
    if (strncmp(addr, "tcp://", 6) == 0) return "TCP";
    if (strncmp(addr, "ipc://", 6) == 0) return "IPC";
    if (strncmp(addr, "inproc://", 9) == 0) return "INPROC";
    return "UNKNOWN";
}

void check_nanomsg_version(void) {
    // nanomsg 没有直接的版本查询API，这里只是示例
    log_message(LOG_INFO, "Nanomsg library initialized");
}

void print_nanomsg_info(void) {
    printf("Nanomsg PAIR Demo v1.0\n");
    printf("Built with nanomsg library\n");
    printf("Supports TCP, IPC, and INPROC transports\n");
}

int safe_send(int sock, const void* buffer, size_t length, int flags) {
    return nn_send(sock, buffer, length, flags);
}

int safe_recv(int sock, void* buffer, size_t length, int flags) {
    return nn_recv(sock, buffer, length, flags);
}